package com.q.springboot.framework.config.redisson;

import lombok.extern.slf4j.Slf4j;

import java.io.Serializable;
import java.time.LocalDateTime;
import java.util.concurrent.CompletableFuture;

@Slf4j
public abstract class RMessageConsumer<T extends Serializable> extends AbstractQueue {

    public void monitorMessage() {
        CompletableFuture.runAsync(this::pastConsumption);
        super.getTopic().addListener(Object.class, (c, m) -> {
            try {
                Object take = super.getMessageQueue().take();
                log.info("rMessage receiveMessage: {}, receiving time {}", take.toString(), LocalDateTime.now());
                this.useMessage((T) take);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
    }

    protected abstract void useMessage(T message);

    public void pastConsumption() {
        while (super.getRDelayedQueue().size() > 0 || super.getMessageQueue().size() > 0) {
            try {
                Object take = super.getMessageQueue().take();
                log.info("rMessage receiveMessage: {}, receiving time {}", take.toString(), LocalDateTime.now());
                this.useMessage((T) take);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

}
